package dyyx.rpc;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.Date;

import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;

class RpcServerHandler extends ChannelInboundHandlerAdapter {

	private Object obj;
	// private Object response;

	public RpcServerHandler(Object obj) {
		super();
		this.obj = obj;

	}

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		RpcRequestInfo methodAndArgs = (RpcRequestInfo) msg;
		Method method = obj.getClass().getMethod(methodAndArgs.getMethodName(), methodAndArgs.getTypes());
		ctx.writeAndFlush(method.invoke(obj, methodAndArgs.getObjects()));
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		System.out.println("channelInactive," + ctx);
	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		System.out.println("channelActive," + ctx);
		
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		System.out.println("exceptionCaught," + cause);
		ctx.close();
	}
}

class RpcClientHander extends ChannelInboundHandlerAdapter {
	private Object response;

	public Object getResponse() {
		return response;
	}

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		response = msg;
		System.out.println("msg from server:" + msg);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		System.out.println("exceptionCaught," + cause);
	}
}

public class RpcUtil {

	/**
	 * 服务注册
	 * 
	 * @param obj
	 *            需要注册的服务实现类
	 * @param port
	 *            端口
	 * @param ip
	 *            地址
	 * @throws InterruptedException
	 */
	public static void regist(final Object obj, int port, String ip) throws InterruptedException {
		int BIZGROUPSIZE = Runtime.getRuntime().availableProcessors() * 2;

		int BIZTHREADSIZE = 100;
		EventLoopGroup bossGroup = new NioEventLoopGroup(BIZGROUPSIZE);
		EventLoopGroup workerGroup = new NioEventLoopGroup(BIZTHREADSIZE);
		if (obj == null)
			throw new IllegalArgumentException("serviceImpl is null");
		if (port <= 0 || port > 65535)
			throw new IllegalArgumentException("port error," + port);
		ServerBootstrap bootstrap = new ServerBootstrap();
		bootstrap.group(bossGroup, workerGroup);
		bootstrap.channel(NioServerSocketChannel.class);
		bootstrap.childHandler(new ChannelInitializer<Channel>() {

			@Override
			protected void initChannel(Channel ch) throws Exception {
				ChannelPipeline pipeline = ch.pipeline();
				pipeline.addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
				pipeline.addLast(new LengthFieldPrepender(4));
				pipeline.addLast("encoder", new ObjectEncoder());
				pipeline.addLast("decoder", new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));

				pipeline.addLast(new RpcServerHandler(obj));
			}
		});
		ChannelFuture f = bootstrap.bind(ip, port).sync();
		f.channel().closeFuture().sync();
		System.out.println("rpc server start,port=" + port + ",now=" + new Date());
	}

	@SuppressWarnings("unchecked")
	public static <T> T getObj(Class<T> interfaceClass, final String host, final int port) {
		if (interfaceClass == null)
			throw new IllegalArgumentException("interfaceClass is null");
		if (!interfaceClass.isInterface())
			throw new IllegalArgumentException(interfaceClass.getName() + " is not interface");
		if (host == null || host.length() == 0)
			throw new IllegalArgumentException("host error");
		if (port <= 0 || port > 65535)
			throw new IllegalArgumentException("port error," + port);

		return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(), new Class<?>[] { interfaceClass },
				new InvocationHandler() {

					@Override
					public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

						RpcRequestInfo mArgs = new RpcRequestInfo(method.getName(), method.getParameterTypes(), args);
						final RpcClientHander tcpClientHander = new RpcClientHander();
						EventLoopGroup group = new NioEventLoopGroup();
						try {
							Bootstrap b = new Bootstrap();
							b.group(group);
							// b.channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY,
							// true);
							b.channel(NioSocketChannel.class).option(ChannelOption.SO_KEEPALIVE, true);

							b.handler(new ChannelInitializer<SocketChannel>() {
								@Override
								protected void initChannel(SocketChannel ch) throws Exception {
									ChannelPipeline pipeline = ch.pipeline();
									pipeline.addLast("frameDecoder",
											new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
									pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));
									pipeline.addLast("encoder", new ObjectEncoder());
									pipeline.addLast("decoder",
											new ObjectDecoder(Integer.MAX_VALUE, ClassResolvers.cacheDisabled(null)));
									pipeline.addLast("handler", tcpClientHander);
								}
							});

							ChannelFuture f = b.connect(host, port).sync();

							f.channel().writeAndFlush(mArgs).sync();
							f.channel().closeFuture().sync();

						} catch (Throwable e) {					
							System.out.println("getObj error,"+e);
							e.printStackTrace();
						} finally {
							group.shutdownGracefully();
						}
						return tcpClientHander.getResponse();

					}
				});
	}
}